package com.lou1052.redis.jedis.source.single;

import com.lou1052.core.base.annotation.thread.ThreadSafe;
import com.lou1052.redis.config.ConfigType;
import com.lou1052.redis.config.NodeConfig;
import com.lou1052.redis.config.NodeInfo;
import com.lou1052.redis.jedis.config.JedisConfig;
import com.lou1052.redis.jedis.config.PoolConfig;
import com.lou1052.redis.jedis.ext.client.single.JedisExt;
import com.lou1052.redis.jedis.ext.client.single.JedisExtPool;
import com.lou1052.redis.jedis.source.AbsJedisSource;
import com.lou1052.redis.jedis.util.SerSelector;
import org.apache.commons.pool2.PooledObject;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.Protocol;

import java.util.Deque;
import java.util.Set;
import java.util.concurrent.BlockingDeque;
import java.util.concurrent.ConcurrentLinkedDeque;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.function.Supplier;

/**
 * single redisExt
 * Created with IntelliJ IDEA.
 * User: liuqing
 * Date: 2015/10/14
 * Time: 22:41
 * To change this template use File | Settings | File Templates.
 */
@ThreadSafe
public class JedisExtSource extends AbsJedisSource<JedisExt> {

    public JedisExtSource(JedisConfig jedisConfig, SerSelector serSelectors,boolean concurrent) {
        super(jedisConfig,serSelectors);
        if(concurrent){
            dequeSupplier = ConcurrentLinkedDeque::new;
        }else{
            dequeSupplier=null;//use default
        }
    }

    private JedisExtPool jedisPool;

    /**
     * 连接池使用的Deque实现
     */
    private final Supplier<Deque<PooledObject<JedisExt>>> dequeSupplier;

    /**
     * 获取Jedis资源
     * 释放资源使用 jedis.close()
     * @return Jedis
     */
    @Override
    public JedisExt getSource(){
        JedisExt jedis = jedisPool.getResource();
        jedis.setSerializerFunction(serializerFunction);
        return jedis;
    }

    @Override
    public void destroy() throws Exception {
        jedisPool.destroy();
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        if(jedisConfig.getConfigType()!= ConfigType.SINGER){
            throw  new IllegalArgumentException("Jedis singer init but config is not ConfigType.SINGER");
        }
        PoolConfig poolConfig = jedisConfig.getPoolConfig();
        NodeConfig nodeConfig = jedisConfig.getNodeConfig();
        Set<NodeInfo> hosts = nodeConfig.getNodeInfoSet();
        if(hosts.size()!=1){
            throw  new IllegalArgumentException("Jedis Singer init, but config is not singer ip hosts.size()="+hosts.size());
        }
        NodeInfo host = (NodeInfo)hosts.toArray()[0];
        String ip = host.getIp();
        int port = host.getPort();
        String passWord = host.getAuth();
        if(passWord==null){
            jedisPool = new JedisExtPool(poolConfig, ip, port,dequeSupplier);
        }else {
            jedisPool = new JedisExtPool(poolConfig, ip, port, Protocol.DEFAULT_TIMEOUT, passWord, Protocol.DEFAULT_DATABASE,dequeSupplier);
        }
        logger.debug("init singer Jedis clients");
    }
}
